package com.test.producer.hello;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Semaphore;

/**
 * @program: spring-boot-learning
 * @description: hello发送端
 * @author: https://niaobulashi.com
 * @create: 2019-05-09 16:22
 **/
@Slf4j
@Component
public class HelloSender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private RedisTemplate redisTemplate;

    Semaphore semaphore = new Semaphore(500);

    public static final String MESSAGE_CONFIRM_KEY = "message_confirm_";

    public HelloSender(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    /**
     * 如果消息没有到达交换机,则该方法中isSendSuccess = false,error为错误信息;
     * 如果消息正确到达交换机,则该方法中isSendSuccess = true;
     */
    @SneakyThrows
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            log.info("成功通知 ：correlationData: {} ,ack:{},cause:{}", correlationData, ack, cause);
            //成功通知
            //删除redis中的相关数据
            if (correlationData != null) {
                redisTemplate.delete(correlationData.getId());
                redisTemplate.delete(MESSAGE_CONFIRM_KEY + correlationData.getId());
            }
        } else {
            //失败通知
            //从redis中获取刚才的消息内容
            Map<String, String> map = (Map<String, String>) redisTemplate.opsForHash().entries(MESSAGE_CONFIRM_KEY + correlationData.getId());
            //重新发送
            String exchange = map.get("exchange");
            String routingkey = map.get("routingkey");
            String message = map.get("message");
            log.info("失败通知 ：correlationData: {} ,ack:{},cause:{}", correlationData, ack, cause);
            log.error("重新发送 ：{}", message);
            rabbitTemplate.convertAndSend(exchange, routingkey, message, correlationData);
        }

    }

    /**
     * 消息从交换机成功到达队列，则returnedMessage方法不会执行;
     * 消息从交换机未能成功到达队列，则returnedMessage方法会执行;
     * //        原文链接：https://blog.csdn.net/Weixiaohuai/article/details/94961012
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error(" returnedMessage 重新发送 ：{}", message);
        rabbitTemplate.convertAndSend(exchange, routingKey, message);
    }

    /**
     * 自定义消息发送方法
     *
     * @param exchange
     * @param routingKey
     * @param message
     */
    @Async
    public void sendMessage(String exchange, String routingKey, String message) {
        System.out.println("当前线程名称" + Thread.currentThread().getName() + "message: " + message);
        try {
            // 信号量控制并发发送数量
//            semaphore.acquire();
            //设置消息的唯一标识并存入到redis中
            CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
            redisTemplate.opsForValue().set(correlationData.getId(), message);

            //将本次发送消息的相关元数据保存到redis中
            Map<String, String> map = new HashMap<>(4);
            map.put("exchange", exchange);
            map.put("routingKey", routingKey);
            map.put("message", message);
            redisTemplate.opsForHash().putAll(MESSAGE_CONFIRM_KEY + correlationData.getId(), map);

            //携带着本次消息的唯一标识,进行数据发送
            rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
        } catch (Exception e) {
            log.error("发送消息异常", e);
            throw new RuntimeException("发送消息异常" + e.getMessage());
        } finally {
//            semaphore.release();
        }

    }


}

